/**
 *  Copyright 2011 Pavlov Dm.
 *  Licensed under the MIT License
 */
 
package org.motiv.store;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.io.File;
import java.io.Serializable;
import java.io.RandomAccessFile;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.ByteArrayInputStream;
import org.apache.log4j.Logger;
import java.util.Iterator;
import java.util.Map;
import java.util.NoSuchElementException;

import org.motiv.core.Element;
import org.motiv.core.CacheException;
import org.motiv.core.Cache;
import org.motiv.config.CacheConfiguration;
import org.motiv.utils.MemoryEfficientByteArrayOutputStream;
import org.motiv.policy.BasePolicy;
/**
 * A disk store implementation.
 * @author Pavlov Dm
 */
public class DiskStore implements Store {

	private static final int ONE_MEGABYTE = 1048576;
	private static final int MS_PER_SECOND = 1000;
	private static final int SPOOL_THREAD_INTERVAL = 200;
	private static final int ESTIMATED_MINIMUM_PAYLOAD_SIZE = 512;
	private static final long MAX_EVICTION_RATIO = 5L;
	public static final String DISK_PATH_DIRECTORY_PREFIX = "l2cache_auto_created";
   
	/** Logger instance */
	private static final Logger LOG=Logger.getLogger(DiskStore.class);
   
    private final String name;
   
    private ConcurrentHashMap diskElements = new ConcurrentHashMap();
	
    private List freeSpace = Collections.synchronizedList(new ArrayList());
	
    private volatile ConcurrentHashMap spool = new ConcurrentHashMap();
	
    private File dataFile;

    private File indexFile;
	
	private Thread spoolAndExpiryThread;
	
    private volatile boolean spoolAndExpiryThreadActive;
	
	private final long expiryThreadInterval = 10;
	
	private volatile boolean active;
	
	private Cache cache;
	
	private final String diskPath;
	 /**
     * The maximum elements to allow in the disk file.
     */
    private volatile long maxElementsOnDisk;
	
	private final int diskSpoolBufferSizeBytes;
	
	private final boolean eternal;
	
	private RandomAccessFile randomAccessFile;
	
	/* Lock around spool field. */
    private final Object spoolLock = new Object();
	/**
     * The size in bytes of the disk elements
     */
    private final AtomicLong totalSize = new AtomicLong();
	
	private volatile int lastElementSize;
	
    /**
     * Creates a disk store.
     *
     */
    public DiskStore(Cache cache) {
		this.cache = cache;
        name = cache.getName();
		diskPath = generateUniqueDirectory();
		CacheConfiguration config = cache.getCacheConfiguration();
		maxElementsOnDisk = config.getMaxElementsOnDisk();
		diskSpoolBufferSizeBytes = config.getDiskSpoolBufferSizeMB() * ONE_MEGABYTE;
		eternal = config.isEternal();
	    try {
			initialiseFiles();
            active = true;
            // Always start up the spool thread
            spoolAndExpiryThread = new SpoolAndExpiryThread();
            spoolAndExpiryThread.start();
        } catch (final Exception e) {
            dispose();
            throw new CacheException(name + "DiskStore: Could not create disk store. Initial cause was " + e.getMessage());
        }
    }
	/**
	* Check directory and create data and index files for store
	*/
	private void initialiseFiles() throws Exception {
        if (diskPath == null) {
          throw new CacheException(cache.getName() + " Cache: Could not create disk store. " +
                  "This CacheManager configuration does not allow creation of DiskStores. " +
                  "If you wish to create DiskStores, please configure a diskStore path.");
        }

        final File diskDir = new File(diskPath);
        // Make sure the cache directory exists
        if (diskDir.exists() && !diskDir.isDirectory()) {
            throw new Exception("Store directory \"" + diskDir.getCanonicalPath() + "\" exists and is not a directory.");
        }
        if (!diskDir.exists() && !diskDir.mkdirs()) {
            throw new Exception("Could not create cache directory \"" + diskDir.getCanonicalPath() + "\".");
        }

        dataFile = new File(diskDir, getDataFileName());
        indexFile = new File(diskDir, getIndexFileName());
		//Delete old files if exists
		if(dataFile.exists()){
			deleteFile(dataFile);
		} 
		if(indexFile.exists()){
			deleteFile(indexFile);
		}
		this.randomAccessFile = new RandomAccessFile(dataFile, "rw");
	}
    /**
     * Gets an {@link Element} from the Disk Store.
     * @return The element
     */
    public final Element get(final Object key) {
        try {
            checkActive();
			//Check element in spool
            synchronized (spoolLock) {
                Element element = (Element) spool.remove(key);
                if (element != null) {
                    return element;
                }
            }
            // Check if the element is on disk
            final DiskElement diskElement = (DiskElement) diskElements.get(key);
            if (diskElement == null) {
                // Not on disk
                return null;
            }
            synchronized (diskElement) {
                if (diskElement.isValid()) {
                    return loadElementFromDiskElement(diskElement);
                } else {
                    return null;
                }
            }

        } catch (Exception exception) {
            LOG.error(name + "Cache: Could not read disk store element for key " + key + ". Error was "+ exception.getMessage());
        }
        return null;
    }
	 private Element loadElementFromDiskElement(DiskElement diskElement) throws IOException, ClassNotFoundException {
        final byte[] buffer = new byte[diskElement.payloadSize];
        synchronized (randomAccessFile) {
            // Load the element
            randomAccessFile.seek(diskElement.position);
            randomAccessFile.readFully(buffer);
        }
        final ByteArrayInputStream instr = new ByteArrayInputStream(buffer);
        final ObjectInputStream objstr = new ObjectInputStream(instr);
		Element element = (Element) objstr.readObject();
		objstr.close();
		return element;
    }
    /**
     * Puts an element into the disk store.
     */
    public final boolean put(final Element element) {
		boolean newPut = !this.containsKey(element.getObjectKey());
        try {
            checkActive();

            // Spool the element
            if (spoolAndExpiryThread.isAlive()) {
                spool.put(element.getObjectKey(), element);
            } else {
                LOG.error(name + "Cache: Elements cannot be written to disk store because the spool thread has died.");
                spool.clear();
            }
            return newPut;
        } catch (Exception e) {
            LOG.error(name + "Cache: Could not write disk store element for " + element.getObjectKey()
                    + ". Initial cause was " + e.getMessage(), e);
            return newPut;
        }
    }
    /**
     * Removes an item from the disk store.
     */
    public final synchronized Element remove(final Object key) {
        try {
            checkActive();
            Element element = (Element) spool.remove(key);
            final DiskElement diskElement = (DiskElement) diskElements.remove(key);
            if (diskElement != null) {
                element = loadElementFromDiskElement(diskElement);
                freeBlock(diskElement);
            }
            return element;
        } catch (Exception exception) {
            String message = name + "Cache: Could not remove disk store entry for key " + key
                    + ". Error was " + exception.getMessage();
            LOG.error(message, exception);
            throw new CacheException(message);
        }
    }
    /**
     * Remove all of the elements from the store.
     */
    public final synchronized void removeAll() {
		try {
            checkActive();
            spool = new ConcurrentHashMap();
            diskElements = new ConcurrentHashMap();
            freeSpace = Collections.synchronizedList(new ArrayList());
            totalSize.set(0L);
            synchronized (randomAccessFile) {
                randomAccessFile.setLength(0);
            }
        } catch (Exception e) {
            LOG.error(name + " Cache: Could not rebuild disk store. Initial cause was " + e.getMessage(), e);
            dispose();
        }
    }
	
	/**
    * A reference to an on-disk elements.
    * <p/>
    * Copies of expiryTime and hitcount are held here as a performance optimisation, so
    * that we do not need to load the data from Disk to get this often used information.
    */
    public static final class DiskElement implements Serializable {
        /**
         * the file pointer
         */
        private long position;
        /**
         * The size used for data.
         */
        private int payloadSize;
        /**
         * the size of this element.
         */
        private int blockSize;
        /**
         * The key this element is mapped with in DiskElements. 
         */
        private Object key;
        /**
         * The expiry time in milliseconds
         */
        private long expiryTime;
        /**
         * The number of times the element has been requested and found in the cache.
         */
        private long hitcount;
        /**
         * @return the key of this object
         */
        public Object getObjectKey() {
            return key;
		}
        /**
         * @return the hit count for the element
         */
        public long getHitCount() {
            return hitcount;
        }
        /**
         * @return the time at which this element will expire
         */
        public long getExpiry() {
            return expiryTime;
        }
        /**
         * @return the size of this element on disk
         */
        public int getSize() {
            return payloadSize;
        }
        /**
         * @return the starting offset of this element on disk
         */
        public long getPosition() {
            return position;
        }
        /**
         * free.
         */
        public void free() {
            this.position = -1;
        }
        /**
         * isValid.
         * @return boolean
         */
        public boolean isValid() {
            return position >= 0;
        }
    }
	
	/**
     * A background daemon thread that writes objects to the file.
     */
    private final class SpoolAndExpiryThread extends Thread {
        public SpoolAndExpiryThread() {
            super("Store " + name + " Spool Thread");
            setDaemon(true);
            setPriority(Thread.NORM_PRIORITY);
            spoolAndExpiryThreadActive = true;
        }
        /**
        * RemoteDebugger thread method.
        */
        @Override
        public final void run() {
            spoolAndExpiryThreadMain();
        }
    }
	
    private void spoolAndExpiryThreadMain() {
		long nextExpiryTime = System.currentTimeMillis();
        while (spoolAndExpiryThreadActive) {
			try{
				Thread.sleep(SPOOL_THREAD_INTERVAL);
            } catch (InterruptedException e) {}
			try{
				flushSpool();
			} catch(IOException ex){}
            if (!spoolAndExpiryThreadActive) {
                return;
            }
			nextExpiryTime = expireElementsIfRequired(nextExpiryTime);
        }
    }
	private long expireElementsIfRequired(long nextExpiryTime) {
        long updatedNextExpiryTime = nextExpiryTime;
        if (!eternal && System.currentTimeMillis() > nextExpiryTime) {
            try {
                updatedNextExpiryTime += expiryThreadInterval * MS_PER_SECOND;
                expireElements();
            } catch (Throwable e) {
                LOG.error(name + " Cache: Could not expire elements from disk due to "
                        + e.getMessage() + ". Continuing...", e);
            }
        }
        return updatedNextExpiryTime;
    }
	/**
     * Removes expired elements.
     */
    public void expireElements() {
        final long now = System.currentTimeMillis();
        // Clean up the spool
        for (Iterator iterator = spool.values().iterator(); iterator.hasNext();) {
            final Element element = (Element) iterator.next();
            if (element.isExpired()) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug(name + "Cache: Removing expired spool element " + element.getObjectKey());
                }
                iterator.remove();
            }
        }
        // Clean up disk elements
        for (Iterator iterator = diskElements.entrySet().iterator(); iterator.hasNext();) {
            final Map.Entry entry = (Map.Entry) iterator.next();
            DiskElement diskElement = (DiskElement) entry.getValue();

            if (now >= diskElement.expiryTime) {
                diskElement = (DiskElement) diskElements.remove(entry.getKey());
                if (diskElement != null) {
                    freeBlock(diskElement);
                }
            }
        }
    }

	private ConcurrentHashMap swapSpoolReference() {
        ConcurrentHashMap copyOfSpool = spool;
        spool = new ConcurrentHashMap();
        return copyOfSpool;
    }
	/**
     * Flushes all spooled elements to disk.
     */
    private synchronized void flushSpool() throws IOException {
        if (spool.size() == 0) {
            return;
        }
        synchronized (spoolLock) {
            ConcurrentHashMap copyOfSpool = swapSpoolReference();
            //does not guarantee insertion order
            Iterator valuesIterator = copyOfSpool.values().iterator();
            while (valuesIterator.hasNext()) {
                writeOrReplaceEntry(valuesIterator.next());
                valuesIterator.remove();
            }
        }
    }
	private void writeOrReplaceEntry(Object object) throws IOException {
        Element element = (Element) object;
        if (element == null) {
            return;
        }
        final Serializable key = (Serializable) element.getObjectKey();
        removeOldEntry(key);
        if (maxElementsOnDisk > 0 && diskElements.size() >= maxElementsOnDisk) {
            long overflow = 1 + diskElements.size() - maxElementsOnDisk;
            evictLfuDiskElements((int) Math.min(overflow, MAX_EVICTION_RATIO));
        }
        writeElement(element, key);
    }
	private void evictLfuDiskElements(int count) {
        for (int i = 0; i < count; i++) {
            DiskElement diskElement = findRelativelyUnused();
            diskElement = (DiskElement) diskElements.remove(diskElement.key);
			if(LOG.isDebugEnabled()){
				LOG.debug("Element with key "+diskElement.key+" evicted");
			}
            if (diskElement != null) {
                freeBlock(diskElement);
            }
        }
    }
	/**
     * Find a "relatively" unused disk element, but not the element just added.
     * @return a DiskElement likely not to be in the bottom quartile of use
     */
    private DiskElement findRelativelyUnused() {
        return leastHit(sampleElements(diskElements), null);
    }
	 /**
     * Finds the least hit of the sampled elements provided
     * @param sampledElements this should be a random subset of the population
     * @param justAdded       we never want to select the element just added. May be null.
     * @return the least hit
     */
    public static DiskElement leastHit(DiskElement[] sampledElements, DiskElement justAdded) {
        //edge condition when Memory Store configured to size 0
        if (sampledElements.length == 1 && justAdded != null) {
            return justAdded;
        }
        DiskElement lowestElement = null;
        for (DiskElement diskElement : sampledElements) {
            if (lowestElement == null) {
                if (!diskElement.equals(justAdded)) {
                    lowestElement = diskElement;
                }
            } else {
                if (diskElement.getHitCount() < lowestElement.getHitCount() && !diskElement.equals(justAdded)) {
                    lowestElement = diskElement;
                }
            }
        }
        return lowestElement;
    }
    /**
     * Uses random numbers to sample the entire map.
     * @param map
     * @return an array of sampled elements
     */
    private DiskElement[] sampleElements(ConcurrentHashMap map) {
        int[] offsets = BasePolicy.generateRandomSample(map.size());
        DiskElement[] elements = new DiskElement[offsets.length];
        Iterator iterator = map.values().iterator();
        for (int i = 0; i < offsets.length; i++) {
            try {
                for (int j = 0; j < offsets[i]; j++) {
                    iterator.next();
                }
                elements[i] = (DiskElement) iterator.next();
            } catch (NoSuchElementException e) {
                DiskElement[] trimmed = new DiskElement[i];
                System.arraycopy(elements, 0, trimmed, 0, i);
                return trimmed;
            }
        }
        return elements;
    }

	 /**
     * Remove the old entry
     * @param key
     */
    private void removeOldEntry(Serializable key) {
        final DiskElement oldBlock = (DiskElement) diskElements.remove(key);
        if (oldBlock != null) {
            freeBlock(oldBlock);
        }
    }
	 /**
     * Marks a block as free.
     * @param diskElement the DiskElement to move to the free space list
     */
    private void freeBlock(final DiskElement diskElement) {
        DiskElement freeBlock = new DiskElement();
        freeBlock.position = diskElement.position;
        freeBlock.blockSize = diskElement.blockSize;
        freeBlock.payloadSize = 0;
        freeBlock.key = null;
        freeBlock.hitcount = 0;
        freeBlock.expiryTime = 0;
        synchronized (diskElement) {
            diskElement.free();
        }
        totalSize.addAndGet(diskElement.payloadSize * -1L);
        freeSpace.add(freeBlock);
    }
	private void writeElement(Element element, Serializable key) throws IOException {
		MemoryEfficientByteArrayOutputStream buffer = MemoryEfficientByteArrayOutputStream.serialize(element, estimatedPayloadSize());
		if (buffer == null) {
            return;
        }
        int bufferLength = buffer.size();
		try {
            DiskElement diskElement = checkForFreeBlock(bufferLength);    
            synchronized (randomAccessFile) {
                randomAccessFile.seek(diskElement.position);
                randomAccessFile.write(buffer.toByteArray(), 0, bufferLength);
            }
            buffer = null;
            // Add to index, update stats
            diskElement.payloadSize = bufferLength;
            diskElement.key = key;
            diskElement.expiryTime = element.getExpirationTime();
            diskElement.hitcount = element.getHitCount();
            totalSize.addAndGet(bufferLength);
            lastElementSize = bufferLength;
            diskElements.put(key, diskElement);
        } catch (OutOfMemoryError e) {
            LOG.error("OutOfMemoryError on serialize: " + key);

        }
    }
	/**
	* Get estimated payload size method
	*/
	private int estimatedPayloadSize() {
        try {
            int size = (int) (totalSize.get() / diskElements.size());
            if (size <= 0) {
                return ESTIMATED_MINIMUM_PAYLOAD_SIZE;
            }
            return size;
        } catch (Exception e) {
            return ESTIMATED_MINIMUM_PAYLOAD_SIZE;
        }
    }
	/**
	* Check free block for element method.
	*/
	private DiskElement checkForFreeBlock(int bufferLength) throws IOException {
        DiskElement diskElement = findFreeBlock(bufferLength);
        if (diskElement == null) {
            diskElement = new DiskElement();
            synchronized (randomAccessFile) {
                diskElement.position = randomAccessFile.length();
            }
            diskElement.blockSize = bufferLength;
        }
        return diskElement;
    }
	 /**
     * Allocates a free block.
     */
    private DiskElement findFreeBlock(final int length) {
        for (int i = 0; i < freeSpace.size(); i++) {
            final DiskElement element = (DiskElement) freeSpace.get(i);
            if (element.blockSize >= length) {
                freeSpace.remove(i);
                return element;
            }
        }
        return null;
    }
	 /**
     * Shuts down the disk store
     */
    public final void dispose() {
        if (!active) return;
        try { 
           
            spoolAndExpiryThreadActive = false;
			
            if (spoolAndExpiryThread != null) {
                spoolAndExpiryThread.interrupt();
                spoolAndExpiryThread.join();
            }

            //Clear in-memory data structures
            spool.clear();
            diskElements.clear();
            freeSpace.clear();
			if (randomAccessFile!= null) {
                synchronized (randomAccessFile) {
                    randomAccessFile.close();
                }
            }
            deleteFilesInDirectory();
        } catch (Exception e) {
            LOG.error(name + "Store: Could not shut down disk store. Initial cause was " + e.getMessage(), e);
        } finally {
            active = false;
            randomAccessFile = null;
            cache = null;
        }
    }
	 /**
     * Get data file name method
     * @return the file name of the data file where the disk store stores data, without any path information.
     */
    public final String getDataFileName() {
        return name + ".data";
    }
	/**
	 * Get index file name method.
     * @return the file name of the index file, which maintains a record of elements and their addresses
     * on the data file, without any path information.
     */
    public final String getIndexFileName() {
        return name + ".index";
    }
	/**
     * Delete file method.
     */
	private static void deleteFile(File f) {
        if (f.delete()) {
            LOG.debug("Deleted file {"+f.getName()+"}");
        } else {
            LOG.debug("Failed to delete file {"+f.getName()+"}");
        }
    }
	 /**
     * Delete files in the auto generated directory. These are one time only for those who create multiple
     * CacheManagers with the same disk path settings.
     */
    protected void deleteFilesInDirectory() {
        if (diskPath != null && !diskPath.isEmpty()) {
            if (dataFile != null && dataFile.exists()) {
				System.out.println("Delete dir: "+diskPath+" "+dataFile.getName());
                LOG.debug("Deleting file " + dataFile.getName());
                deleteFile(dataFile);
            }
            if (indexFile != null && indexFile.exists()) {
                LOG.debug("Deleting file " + indexFile.getName());
                deleteFile(indexFile);
            }
            File dataDirectory = new File(diskPath);
			
            if (dataDirectory != null && dataDirectory.exists()) {
                if (dataDirectory.delete()) {
                    LOG.debug("Deleted directory " + dataDirectory.getName());
                }
            }

        }
    }
	/**
     * Generates a unique directory name for use in automatically creating a diskStorePath where there is a conflict.
     */
    public static String generateUniqueDirectory() {
        return System.getProperty("user.home")+File.separator+DiskStore.DISK_PATH_DIRECTORY_PREFIX + "_" + System.currentTimeMillis()+File.separator;
    }
	 /**
     * Asserts that the store is active.
     */
    private void checkActive() throws CacheException {
        if (!active) {
            throw new CacheException(name + " Cache: The Disk store is not active.");
        }
    }
	 /**
     * Check to see if a key is in the Store.
     * @param key The Element key
     * @return true if found. 
     */
    public final boolean containsKey(Object key) {
        return diskElements.containsKey(key) || spool.containsKey(key);
    }

}
